#include "containerstore.h"
#include "../utils/serial.h"
#include "../utils/sync_queue.h"
#include "../jcr.h"
#include "../destor.h"
#include "../index/kvstore.h"
//#include "../index/simi_detection.h"

extern struct chunk *retrive_basechunk_by_fp_id(fingerprint *fp, containerid id);
static int64_t container_count = 0;
static FILE *fp;
/* Control the concurrent accesses to fp. */
static pthread_mutex_t mutex;

static pthread_t append_t;

static SyncQueue *container_buffer;

struct metaEntry
{
	int32_t off;
	int32_t len;
	fingerprint fp;
	/*
	 * the flag indicates whether it is a delta
	 * 0 -> base chunk
	 * 1 -> delta chunk
	 * */
	char flag;
	uint64_t sf1;
	uint64_t sf2;
	uint64_t sf3;
	struct metaEntry *pre;
	struct metaEntry *next;
	// containerid baseid;
	// int32_t delta_size;
	// int32_t base_size;
};

extern struct index_item
{
	fingerprint fp;
	containerid id;
};

/*
 * We must ensure a container is either in the buffer or written to disks.
 */
static void *append_thread(void *arg)
{

	while (1)
	{
		struct container *c = sync_queue_get_top(container_buffer);
		if (c == NULL)
			break;

		TIMER_DECLARE(1);
		TIMER_BEGIN(1);

		write_container(c);

		TIMER_END(1, jcr.write_time);

		sync_queue_pop(container_buffer);

		free_container(c);
	}

	return NULL;
}

void init_container_store()
{
	sds containerfile = sdsdup(destor.working_directory);
	containerfile = sdscat(containerfile, "/container.pool");

	if ((fp = fopen(containerfile, "r+")))
	{
		fread(&container_count, 8, 1, fp);
	}
	else if (!(fp = fopen(containerfile, "w+")))
	{
		perror(
			"Can not create containers/container.pool for read and write because");
		exit(1);
	}

	sdsfree(containerfile);

	container_buffer = sync_queue_new(25);

	pthread_mutex_init(&mutex, NULL);

	pthread_create(&append_t, NULL, append_thread, NULL);
}

void close_container_store()
{
	sync_queue_term(container_buffer);
	pthread_join(append_t, NULL);

	fseek(fp, 0, SEEK_SET);
	fwrite(&container_count, sizeof(container_count), 1, fp);

	fclose(fp);
	fp = NULL;

	pthread_mutex_destroy(&mutex);
}

static void init_container_meta(struct containerMeta *meta)
{
	meta->chunk_num = 0;
	meta->data_size = 0;
	meta->id = TEMPORARY_ID;
	meta->map = g_hash_table_new_full(g_int_hash, g_fingerprint_equal, NULL,
									  free);
	meta->delta_sf1 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, NULL, NULL);
	meta->delta_sf2 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, NULL, NULL);
	meta->delta_sf3 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, NULL, NULL);
	meta->head = meta->tail = NULL;
}

/*
 * For backup.
 */
struct container *create_container()
{
	struct container *c = (struct container *)malloc(sizeof(struct container));
	c->data = calloc(1, CONTAINER_SIZE);

	init_container_meta(&c->meta);
	c->meta.id = container_count++;
	return c;
}

containerid get_container_id(struct container *c)
{
	return c->meta.id;
}

void write_container_async(struct container *c)
{
	assert(c->meta.chunk_num == g_hash_table_size(c->meta.map));

	if (container_empty(c))
	{
		/* An empty container
		 * It possibly occurs in the end of backup */
		container_count--;
		VERBOSE("Append phase: Deny writing an empty container %lld",
				c->meta.id);
		return;
	}
	sync_queue_push(container_buffer, c);

	jcr.container_num_now_stored = c->meta.id;
}

/*
 * Called by Append phase
 * HEAD, meta entries, bitmap, data
 */
void write_container(struct container *c)
{

	assert(c->meta.chunk_num == g_hash_table_size(c->meta.map));

	if (container_empty(c))
	{
		/* An empty container
		 * It possibly occurs in the end of backup */
		container_count--;
		VERBOSE("Append phase: Deny writing an empty container %lld",
				c->meta.id);
		return;
	}

	VERBOSE("Append phase: Writing container %lld of %d chunks", c->meta.id,
			c->meta.chunk_num);

	unsigned char *cur = &c->data[CONTAINER_SIZE - CONTAINER_META_SIZE];
	ser_declare;
	ser_begin(cur, CONTAINER_META_SIZE);
	ser_int64(c->meta.id);
	ser_int32(c->meta.chunk_num);
	ser_int32(c->meta.data_size);

	/*
	GHashTableIter iter;
	gpointer key, value;
	g_hash_table_iter_init(&iter, c->meta.map);
	while (g_hash_table_iter_next(&iter, &key, &value))
	{
		struct metaEntry *me = (struct metaEntry *)value;
		ser_bytes(&me->fp, sizeof(fingerprint));
		ser_bytes(&me->len, sizeof(int32_t));
		ser_bytes(&me->off, sizeof(int32_t));
		ser_bytes(&me->flag, sizeof(char));
		ser_bytes(&me->sf1, sizeof(uint64_t));
		ser_bytes(&me->sf2, sizeof(uint64_t));
		ser_bytes(&me->sf3, sizeof(uint64_t));
	}
	*/
	struct metaEntry *me = (c->meta).head;
	while (me)
	{
		ser_bytes(&me->fp, sizeof(fingerprint));
		ser_bytes(&me->len, sizeof(int32_t));
		ser_bytes(&me->off, sizeof(int32_t));
		ser_bytes(&me->flag, sizeof(char));
		if (me->flag == 0)
		{
			ser_bytes(&me->sf1, sizeof(uint64_t));
			ser_bytes(&me->sf2, sizeof(uint64_t));
			ser_bytes(&me->sf3, sizeof(uint64_t));
		}
		me = me->next;
	}

	ser_end(cur, CONTAINER_META_SIZE);

	pthread_mutex_lock(&mutex);

	if (fseek(fp, c->meta.id * CONTAINER_SIZE + 8, SEEK_SET) != 0)
	{
		perror("Fail seek in container store.");
		exit(1);
	}
	if (fwrite(c->data, CONTAINER_SIZE, 1, fp) != 1)
	{
		perror("Fail to write a container in container store.");
		exit(1);
	}

	pthread_mutex_unlock(&mutex);

	jcr.container_num_now_written = c->meta.id;
}

struct container *retrieve_container_by_id(containerid id)
{
	struct container *c = (struct container *)malloc(sizeof(struct container));
	init_container_meta(&c->meta);
	c->data = malloc(CONTAINER_SIZE);

	pthread_mutex_lock(&mutex);

	fseek(fp, id * CONTAINER_SIZE + 8, SEEK_SET);
	fread(c->data, CONTAINER_SIZE, 1, fp);

	pthread_mutex_unlock(&mutex);

	unsigned char *cur = 0;
	cur = &c->data[CONTAINER_SIZE - CONTAINER_META_SIZE];

	unser_declare;
	unser_begin(cur, CONTAINER_META_SIZE);

	unser_int64(c->meta.id);
	unser_int32(c->meta.chunk_num);
	unser_int32(c->meta.data_size);

	if (c->meta.id != id)
	{
		WARNING("expect %lld, but read %lld", id, c->meta.id);
		assert(c->meta.id == id);
	}

	int i;
	struct metaEntry *last_me = NULL;
	for (i = 0; i < c->meta.chunk_num; i++)
	{
		struct metaEntry *me = (struct metaEntry *)malloc(
			sizeof(struct metaEntry));
		me->pre = me->next = NULL;
		unser_bytes(&me->fp, sizeof(fingerprint));
		unser_bytes(&me->len, sizeof(int32_t));
		unser_bytes(&me->off, sizeof(int32_t));
		unser_bytes(&me->flag, sizeof(char));
		if (me->flag == 0)
		{
			unser_bytes(&me->sf1, sizeof(uint64_t));
			unser_bytes(&me->sf2, sizeof(uint64_t));
			unser_bytes(&me->sf3, sizeof(uint64_t));
		}
		if (i == 0)
		{
			c->meta.head = me;
		}
		else
		{
			last_me->next = me;
			me->pre = last_me;
		}
		last_me = me;
		g_hash_table_insert(c->meta.map, &me->fp, me);
		if (me->flag == 0)
		{
			g_hash_table_replace(c->meta.delta_sf1, &me->sf1, me);
			g_hash_table_replace(c->meta.delta_sf2, &me->sf2, me);
			g_hash_table_replace(c->meta.delta_sf3, &me->sf3, me);
		}
	}
	c->meta.tail = last_me;

	unser_end(cur, CONTAINER_META_SIZE);
	return c;
}

static struct containerMeta *container_meta_duplicate(struct container *c)
{
	struct containerMeta *base = &c->meta;
	struct containerMeta *dup = (struct containerMeta *)malloc(
		sizeof(struct containerMeta));
	init_container_meta(dup);
	dup->id = base->id;
	dup->chunk_num = base->chunk_num;
	dup->data_size = base->data_size;

	struct metaEntry *cur_me = base->head;
	struct metaEntry *last_me = NULL;
	while (cur_me)
	{
		struct metaEntry *me = (struct metaEntry *)malloc(
			sizeof(struct metaEntry));
		memcpy(me, cur_me, sizeof(struct metaEntry));
		me->pre = me->next = NULL;
		if (cur_me == base->head)
			dup->head = me;
		else
		{
			last_me->next = me;
			me->pre = last_me;
		}
		last_me = me;
		g_hash_table_insert(dup->map, &me->fp, me);
		if (me->flag == 0)
		{
			g_hash_table_replace(dup->delta_sf1, &me->sf1, me);
			g_hash_table_replace(dup->delta_sf2, &me->sf2, me);
			g_hash_table_replace(dup->delta_sf3, &me->sf3, me);
		}
		cur_me = cur_me->next;
	}
	dup->tail = last_me;

	return dup;
}

struct index_item *retrieve_fp_and_id(struct containerMeta *cm, struct chunk *ck)
{
	struct index_item *item = (struct index_item *)malloc(sizeof(struct index_item));
	item->id = cm->id;
	struct metaEntry *me = NULL;
	struct metaEntry *me1 = g_hash_table_lookup(cm->delta_sf1, &ck->superfeature->sf1);
	struct metaEntry *me2 = g_hash_table_lookup(cm->delta_sf2, &ck->superfeature->sf2);
	struct metaEntry *me3 = g_hash_table_lookup(cm->delta_sf3, &ck->superfeature->sf3);

	if (me1)
		me = me1;
	else if (me2)
		me = me2;
	else if (me3)
		me = me3;
	assert(me);

	memcpy(item->fp, me->fp, sizeof(fingerprint));
	return item;
}

struct container *container_duplicate(struct container *c)
{
	struct container *base = c;
	struct container *dup = (struct container *)malloc(sizeof(struct container));
	dup->data = calloc(1, CONTAINER_SIZE);
	init_container_meta(&dup->meta);

	dup->meta.id = base->meta.id;
	dup->meta.chunk_num = base->meta.chunk_num;
	dup->meta.data_size = base->meta.data_size;

	struct metaEntry *cur_me = base->meta.head;
	struct metaEntry *last_me = NULL;
	while (cur_me)
	{
		struct metaEntry *me = (struct metaEntry *)malloc(
			sizeof(struct metaEntry));
		memcpy(me, cur_me, sizeof(struct metaEntry));
		me->pre = me->next = NULL;
		if (cur_me == base->meta.head)
			dup->meta.head = me;
		else
		{
			last_me->next = me;
			me->pre = last_me;
		}
		last_me = me;
		g_hash_table_insert(dup->meta.map, &me->fp, me);
		if (me->flag == 0)
		{
			g_hash_table_replace(dup->meta.delta_sf1, &me->sf1, me);
			g_hash_table_replace(dup->meta.delta_sf2, &me->sf2, me);
			g_hash_table_replace(dup->meta.delta_sf3, &me->sf3, me);
		}
		cur_me = cur_me->next;
	}
	dup->meta.tail = last_me;

	memcpy(dup->data, base->data, CONTAINER_SIZE);

	return dup;
}

struct container *retrieve_container_in_write_buffer(containerid id)
{
	struct container *con = NULL;
	con = sync_queue_find(container_buffer, container_check_id, &id,
						  container_duplicate);
	if (con)
		return con;
}

struct containerMeta *retrieve_container_meta_by_id(containerid id)
{
	struct containerMeta *cm = NULL;

	/* First, we find it in the buffer */

	cm = sync_queue_find(container_buffer, container_check_id, &id,
						 container_meta_duplicate);

	if (cm)
		return cm;

	cm = (struct containerMeta *)malloc(sizeof(struct containerMeta));
	init_container_meta(cm);

	unsigned char buf[CONTAINER_META_SIZE];

	pthread_mutex_lock(&mutex);

	if (destor.simulation_level >= SIMULATION_APPEND)
		fseek(fp, id * CONTAINER_META_SIZE + 8, SEEK_SET);
	else
		fseek(fp, (id + 1) * CONTAINER_SIZE - CONTAINER_META_SIZE + 8,
			  SEEK_SET);

	fread(buf, CONTAINER_META_SIZE, 1, fp);

	pthread_mutex_unlock(&mutex);

	unser_declare;
	unser_begin(buf, CONTAINER_META_SIZE);

	unser_int64(cm->id);
	unser_int32(cm->chunk_num);
	unser_int32(cm->data_size);

	if (cm->id != id)
	{
		WARNING("expect %lld, but read %lld", id, cm->id);
		assert(cm->id == id);
	}

	int i;
	struct metaEntry *last_me = NULL;
	for (i = 0; i < cm->chunk_num; i++)
	{
		struct metaEntry *me = (struct metaEntry *)malloc(
			sizeof(struct metaEntry));
		me->pre = me->next = NULL;
		unser_bytes(&me->fp, sizeof(fingerprint));
		unser_bytes(&me->len, sizeof(int32_t));
		unser_bytes(&me->off, sizeof(int32_t));
		unser_bytes(&me->flag, sizeof(char));
		if (me->flag == 0)
		{
			unser_bytes(&me->sf1, sizeof(uint64_t));
			unser_bytes(&me->sf2, sizeof(uint64_t));
			unser_bytes(&me->sf3, sizeof(uint64_t));
		}

		if (i == 0)
		{
			cm->head = me;
		}
		else
		{
			last_me->next = me;
			me->pre = last_me;
		}
		last_me = me;
		g_hash_table_insert(cm->map, &me->fp, me);
		if (me->flag == 0)
		{
			g_hash_table_replace(cm->delta_sf1, &me->sf1, me);
			g_hash_table_replace(cm->delta_sf2, &me->sf2, me);
			g_hash_table_replace(cm->delta_sf3, &me->sf3, me);
		}
	}
	cm->tail = last_me;
	return cm;
}

static struct metaEntry *get_metaentry_in_container_meta(struct containerMeta *cm, fingerprint *fp)
{
	return g_hash_table_lookup(cm->map, fp);
}

struct chunk *get_chunk_in_container(struct container *c, fingerprint *fp)
{
	struct metaEntry *me = get_metaentry_in_container_meta(&c->meta, fp);

	assert(me);

	struct chunk *ck = NULL;
	if (me->flag)
	{
		/* It is a delta */
		int32_t csize = 0;
		int32_t delta_size = me->len - sizeof(csize) - sizeof(containerid) - sizeof(fingerprint);

		unser_declare;
		unser_begin(c->data + me->off, 0);
		unser_bytes(&csize, sizeof(int32_t));
		ck = new_chunk(csize);

		ck->delta = new_delta(delta_size);
		unser_bytes(ck->delta->data, delta_size);
		unser_bytes(&ck->delta->baseid, sizeof(containerid));
		unser_bytes(ck->delta->basefp, sizeof(fingerprint));
		unser_end(c->data + me->off, me->len);

		/* The data portion remains unknown */
	}
	else
	{
		/* It is not a delta */
		ck = new_chunk(me->len);
		ck->size = me->len;
		memcpy(ck->data, c->data + me->off, me->len);
	}

	ck->id = c->meta.id;
	// memcpy(&ck->fp, &fp, sizeof(fingerprint)); changed by zyc 2015.7.22
	memcpy(ck->fp, fp, sizeof(fingerprint));

	return ck;
}

struct chunk *get_basechunk_in_container(struct container *c, fingerprint *fp)
{
	struct metaEntry *me = get_metaentry_in_container_meta(&c->meta, fp);

	assert(me);

	if (me->flag != 0)
	{
		unsigned char buf[41];
		hash2code(fp, buf);
		buf[40] = 0;
		printf("fp: %s\n", buf);
	}
	assert(me->flag == 0);

	struct chunk *ck = new_chunk(me->len);
	memcpy(ck->data, c->data + me->off, me->len);

	ck->id = c->meta.id;
	ck->size = me->len;
	memcpy(ck->fp, fp, sizeof(fingerprint));

	return ck;
}

int container_overflow(struct container *c, struct chunk *ck)
{
	int size = ck->size;
	if (c->meta.data_size + size > CONTAINER_SIZE - CONTAINER_META_SIZE)
		return 1;
	/*
	 * 53 is the size of metaEntry.
	 */
	if ((c->meta.chunk_num + 1) * 53 + 16 > CONTAINER_META_SIZE)
		return 1;
	return 0;
}

/*
 * For backup.
 * return 1 indicates success.
 * return 0 indicates fail.
 */
int add_chunk_to_container(struct container *c, struct chunk *ck)
{
	assert(!container_overflow(c, ck));
	if (g_hash_table_contains(c->meta.map, &ck->fp))
	{
		// NOTICE("Writing a chunk already in the container buffer!");
		ck->id = c->meta.id;
		return 0;
	}

	struct metaEntry *me = (struct metaEntry *)malloc(sizeof(struct metaEntry));
	memcpy(me->fp, ck->fp, sizeof(fingerprint));
	me->off = c->meta.data_size;

	/*
	unsigned char buf[41];
	hash2code(ck->fp, buf);
	buf[40] = 0;
	if (buf[0] == 'D' && buf[1] == 'F' && buf[2] == '2'
			&& buf[3] == 'B' && buf[4] == 'A' && buf[5] == 'A') {
		jcr.zero_chunk_num++;
		printf("fp: %s\n", buf);
	}
	*/

	me->sf1 = ck->superfeature->sf1;
	me->sf2 = ck->superfeature->sf2;
	me->sf3 = ck->superfeature->sf3;
	me->pre = me->next = NULL;

	if (!ck->delta)
	{

		me->flag = 0;
		memcpy(c->data + c->meta.data_size, ck->data, ck->size);
		me->len = ck->size;

		c->meta.data_size += me->len;

		if (c->meta.head == NULL)
		{
			c->meta.head = me;
			c->meta.tail = me;
		}
		else
		{
			assert(c->meta.tail != NULL);
			c->meta.tail->next = me;
			me->pre = c->meta.tail;
			c->meta.tail = me;
		}

		g_hash_table_insert(c->meta.map, &me->fp, me);
		c->meta.chunk_num++;

		ck->id = c->meta.id;
	}
	else
	{
		jcr.base_chunk_referred++;

		me->flag = 1;

		if (ck->delta->baseid <= jcr.container_num_of_first_backup)
			jcr.base_chunk_num_in_ver1++;

		ser_declare;
		ser_begin(c->data + c->meta.data_size, 0);
		ser_bytes(&ck->size, sizeof(int32_t));
		ser_bytes(ck->delta->data, ck->delta->size);
		ser_bytes(&ck->delta->baseid, sizeof(containerid));
		ser_bytes(ck->delta->basefp, sizeof(fingerprint));
		ser_end(c->data + c->meta.data_size, sizeof(int32_t) + ck->delta->size + sizeof(containerid) + sizeof(fingerprint));
		me->len = ser_length(c->data + c->meta.data_size);

		c->meta.data_size += me->len;

		if (c->meta.head == NULL)
		{
			c->meta.head = me;
			c->meta.tail = me;
		}
		else
		{
			assert(c->meta.tail != NULL);
			c->meta.tail->next = me;
			me->pre = c->meta.tail;
			c->meta.tail = me;
		}

		g_hash_table_insert(c->meta.map, &me->fp, me);
		c->meta.chunk_num++;

		ck->id = c->meta.id;
		jcr.delta_compression_times++;
	}

	return 1;
}

void free_container_meta(struct containerMeta *cm)
{
	g_hash_table_destroy(cm->map);
	g_hash_table_destroy(cm->delta_sf1);
	g_hash_table_destroy(cm->delta_sf2);
	g_hash_table_destroy(cm->delta_sf3);
	free(cm);
}

void free_container(struct container *c)
{
	g_hash_table_destroy(c->meta.map);
	g_hash_table_destroy(c->meta.delta_sf1);
	g_hash_table_destroy(c->meta.delta_sf2);
	g_hash_table_destroy(c->meta.delta_sf3);
	if (c->data)
		free(c->data);
	free(c);
}

int container_empty(struct container *c)
{
	return c->meta.chunk_num == 0 ? 1 : 0;
}

/*
 * Return 0 if doesn't exist.
 */
int lookup_fingerprint_in_container_meta(struct containerMeta *cm,
										 fingerprint *fp)
{
	return g_hash_table_lookup(cm->map, fp) == NULL ? 0 : 1;
}

int lookup_sf_in_container_meta(struct containerMeta *cm,
								struct chunk *ck)
{
	struct metaEntry *me1 = g_hash_table_lookup(cm->delta_sf1, &ck->superfeature->sf1);
	struct metaEntry *me2 = g_hash_table_lookup(cm->delta_sf2, &ck->superfeature->sf2);
	struct metaEntry *me3 = g_hash_table_lookup(cm->delta_sf3, &ck->superfeature->sf3);
	if (!me1 && !me2 && !me3)
		return 0;
	return 1;
}

int lookup_fingerprint_in_container(struct container *c, fingerprint *fp)
{
	return lookup_fingerprint_in_container_meta(&c->meta, fp);
}

/*
 * Return 0 if doesn't exist. And get the type of chunk
 */
int lookup_chunk_in_container(struct container *c, struct chunk *ck)
{
	struct containerMeta *cm = &c->meta;
	struct metaEntry *me = g_hash_table_lookup(cm->map, &ck->fp);
	if (me)
	{
		ck->type = me->flag;
		return 1;
	}
	return 0;
}

gint g_container_cmp_desc(struct container *c1, struct container *c2,
						  gpointer user_data)
{
	return g_container_meta_cmp_desc(&c1->meta, &c2->meta, user_data);
}

gint g_container_meta_cmp_desc(struct containerMeta *cm1,
							   struct containerMeta *cm2, gpointer user_data)
{
	return cm2->id - cm1->id;
}

int container_check_id(struct container *c, containerid *id)
{
	return container_meta_check_id(&c->meta, id);
}

int container_meta_check_id(struct containerMeta *cm, containerid *id)
{
	return cm->id == *id ? 1 : 0;
}

/*
 * foreach the fingerprints in the container.
 * Apply the 'func' for each fingerprint.
 */
void container_meta_foreach(struct containerMeta *cm, void (*func)(fingerprint *, void *), void *data)
{
	GHashTableIter iter;
	gpointer key, value;
	g_hash_table_iter_init(&iter, cm->map);
	while (g_hash_table_iter_next(&iter, &key, &value))
	{
		func(key, data);
	}
}

int64_t get_container_count()
{
	return container_count;
}
